package wiki.xsx.core.handler;

import org.springframework.data.redis.connection.MessageListener;
import org.springframework.data.redis.connection.Subscription;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.StringRedisTemplate;
import wiki.xsx.core.util.ConvertUtil;

import java.util.ArrayList;
import java.util.Collection;
import java.util.List;
import java.util.stream.Collectors;

/**
 * 发布与订阅助手
 * @author xsx
 * @date 2019/5/21
 * @since 1.8
 */
public class PubSubHandler {
    /**
     * 对象模板
     */
    private RedisTemplate<String, Object> redisTemplate;
    /**
     * 字符串模板
     */
    private StringRedisTemplate stringRedisTemplate;

    /**
     * 数据库助手构造
     * @param redisTemplate 对象模板
     * @param stringRedisTemplate 字符串模板
     */
    private PubSubHandler(RedisTemplate<String, Object> redisTemplate, StringRedisTemplate stringRedisTemplate) {
        this.redisTemplate = redisTemplate;
        this.stringRedisTemplate = stringRedisTemplate;
    }

    /**
     * 获取实例
     * @param redisTemplate 对象模板
     * @param stringRedisTemplate 字符串模板
     * @return 返回实例
     */
    public static PubSubHandler getInstance(RedisTemplate<String, Object> redisTemplate, StringRedisTemplate stringRedisTemplate) {
        return new PubSubHandler(redisTemplate, stringRedisTemplate);
    }

    /**
     * 获取实例
     * @param dbIndex 数据库索引
     * @return 返回实例
     */
    public static PubSubHandler getInstance(int dbIndex) {
        List<RedisTemplate> templateList = HandlerManager.createTemplate(dbIndex);
        return getInstance(templateList.get(0), (StringRedisTemplate) templateList.get(1));
    }

    /**
     * 发布对象频道
     * @see <a href="http://redis.io/commands/publish">Redis Documentation: PUBLISH</a>
     * @since redis 2.0.0
     * @param channel 频道
     * @param message 对象消息
     * @return 返回收到对象消息的订阅者数量
     */
    public void publishAsObj(String channel, Object message) {
        this.redisTemplate.convertAndSend(channel, message);
    }

    /**
     * 发布字符串频道
     * @see <a href="http://redis.io/commands/publish">Redis Documentation: PUBLISH</a>
     * @since redis 2.0.0
     * @param channel 频道
     * @param message 字符串消息
     */
    public void publish(String channel, String message) {
        this.stringRedisTemplate.convertAndSend(channel, message);
    }

    /**
     * 订阅对象频道
     * @see <a href="http://redis.io/commands/subscribe">Redis Documentation: SUBSCRIBE</a>
     * @since redis 2.0.0
     * @param listener 监听器
     * @param channels 频道
     */
    public void subscribeAsObj(MessageListener listener, String ...channels) {
        this.redisTemplate
            .getRequiredConnectionFactory()
            .getConnection()
            .subscribe(listener, ConvertUtil.toByteArray(this.redisTemplate.getKeySerializer(), channels));
    }

    /**
     * 订阅字符串频道
     * @see <a href="http://redis.io/commands/subscribe">Redis Documentation: SUBSCRIBE</a>
     * @since redis 2.0.0
     * @param listener 监听器
     * @param channels 频道
     */
    public void subscribe(MessageListener listener, String ...channels) {
        this.stringRedisTemplate
            .getRequiredConnectionFactory()
            .getConnection()
            .subscribe(listener, ConvertUtil.toByteArray(this.stringRedisTemplate.getKeySerializer(), channels));
    }

    /**
     * 订阅给定模式的对象频道
     * @see <a href="http://redis.io/commands/psubscribe">Redis Documentation: PSUBSCRIBE</a>
     * @since redis 2.0.0
     * @param listener 监听器
     * @param patterns 模式
     */
    public void pSubscribeAsObj(MessageListener listener, String ...patterns) {
        this.redisTemplate
            .getRequiredConnectionFactory()
            .getConnection()
            .pSubscribe(listener, ConvertUtil.toByteArray(this.redisTemplate.getKeySerializer(), patterns));
    }

    /**
     * 订阅给定模式的字符串频道
     * @see <a href="http://redis.io/commands/psubscribe">Redis Documentation: PSUBSCRIBE</a>
     * @since redis 2.0.0
     * @param listener 监听器
     * @param patterns 模式
     */
    public void pSubscribe(MessageListener listener, String ...patterns) {
        this.stringRedisTemplate
            .getRequiredConnectionFactory()
            .getConnection()
            .pSubscribe(listener, ConvertUtil.toByteArray(this.stringRedisTemplate.getKeySerializer(), patterns));
    }

    /**
     * 添加对象订阅频道
     * @since redis 2.0.0
     * @param channels 频道
     */
    public void addSubChannelsAsObj(String ...channels) {
        Subscription subscription = this.redisTemplate.getRequiredConnectionFactory().getConnection().getSubscription();
        if (subscription!=null&&subscription.isAlive()) {
            subscription.subscribe(ConvertUtil.toByteArray(this.redisTemplate.getKeySerializer(), channels));
        }
    }

    /**
     * 添加字符串订阅频道
     * @since redis 2.0.0
     * @param channels 频道
     */
    public void addSubChannels(String ...channels) {
        Subscription subscription = this.stringRedisTemplate.getRequiredConnectionFactory().getConnection().getSubscription();
        if (subscription!=null&&subscription.isAlive()) {
            subscription.subscribe(ConvertUtil.toByteArray(this.stringRedisTemplate.getKeySerializer(), channels));
        }
    }

    /**
     * 添加给定模式对象订阅频道
     * @since redis 2.0.0
     * @param patterns 模式
     */
    public void addPSubChannelsAsObj(String ...patterns) {
        Subscription subscription = this.redisTemplate.getRequiredConnectionFactory().getConnection().getSubscription();
        if (subscription!=null&&subscription.isAlive()) {
            subscription.pSubscribe(ConvertUtil.toByteArray(this.redisTemplate.getKeySerializer(), patterns));
        }
    }

    /**
     * 添加给定模式字符串订阅频道
     * @since redis 2.0.0
     * @param patterns 模式
     */
    public void addPSubChannels(String ...patterns) {
        Subscription subscription = this.stringRedisTemplate.getRequiredConnectionFactory().getConnection().getSubscription();
        if (subscription!=null&&subscription.isAlive()) {
            subscription.pSubscribe(ConvertUtil.toByteArray(this.stringRedisTemplate.getKeySerializer(), patterns));
        }
    }

    /**
     * 退订所有对象频道
     * @see <a href="http://redis.io/commands/unsubscribe">Redis Documentation: UNSUBSCRIBE</a>
     * @since redis 2.0.0
     */
    public void unsubscribeAsObj() {
        this.unsubscribeAsObj((String[]) null);
    }

    /**
     * 退订所有字符串频道
     * @see <a href="http://redis.io/commands/unsubscribe">Redis Documentation: UNSUBSCRIBE</a>
     * @since redis 2.0.0
     */
    public void unsubscribe() {
        this.unsubscribe((String[]) null);
    }

    /**
     * 退订对象频道
     * @see <a href="http://redis.io/commands/unsubscribe">Redis Documentation: UNSUBSCRIBE</a>
     * @since redis 2.0.0
     * @param channels 频道
     */
    public void unsubscribeAsObj(String ...channels) {
        Subscription subscription = this.redisTemplate.getRequiredConnectionFactory().getConnection().getSubscription();
        if (subscription!=null&&subscription.isAlive()) {
            if (channels!=null) {
                subscription.unsubscribe(ConvertUtil.toByteArray(this.redisTemplate.getKeySerializer(), channels));
            }else {
                subscription.unsubscribe();
            }
        }
    }

    /**
     * 退订字符串频道
     * @see <a href="http://redis.io/commands/unsubscribe">Redis Documentation: UNSUBSCRIBE</a>
     * @since redis 2.0.0
     * @param channels 频道
     */
    public void unsubscribe(String ...channels) {
        Subscription subscription = this.stringRedisTemplate.getRequiredConnectionFactory().getConnection().getSubscription();
        if (subscription!=null&&subscription.isAlive()) {
            if (channels!=null) {
                subscription.unsubscribe(ConvertUtil.toByteArray(this.stringRedisTemplate.getKeySerializer(), channels));
            }else {
                subscription.unsubscribe();
            }
        }
    }

    /**
     * 退订给定模式对象频道
     * @see <a href="http://redis.io/commands/punsubscribe">Redis Documentation: PUNSUBSCRIBE</a>
     * @since redis 2.0.0
     */
    public void pUnsubscribeAsObj() {
        this.pUnsubscribe((String[]) null);
    }

    /**
     * 退订给定模式字符串频道
     * @see <a href="http://redis.io/commands/punsubscribe">Redis Documentation: PUNSUBSCRIBE</a>
     * @since redis 2.0.0
     */
    public void pUnsubscribe() {
        this.pUnsubscribe((String[]) null);
    }

    /**
     * 退订给定模式对象频道
     * @see <a href="http://redis.io/commands/punsubscribe">Redis Documentation: PUNSUBSCRIBE</a>
     * @since redis 2.0.0
     * @param patterns 模式
     */
    public void pUnsubscribeAsObj(String ...patterns) {
        Subscription subscription = this.redisTemplate.getRequiredConnectionFactory().getConnection().getSubscription();
        if (subscription!=null&&subscription.isAlive()) {
            if (patterns!=null) {
                subscription.pUnsubscribe(ConvertUtil.toByteArray(this.redisTemplate.getKeySerializer(), patterns));
            }else {
                subscription.pUnsubscribe();
            }
        }
    }

    /**
     * 退订给定模式字符串频道
     * @see <a href="http://redis.io/commands/punsubscribe">Redis Documentation: PUNSUBSCRIBE</a>
     * @since redis 2.0.0
     * @param patterns 模式
     */
    public void pUnsubscribe(String ...patterns) {
        Subscription subscription = this.stringRedisTemplate.getRequiredConnectionFactory().getConnection().getSubscription();
        if (subscription!=null&&subscription.isAlive()) {
            if (patterns!=null) {
                subscription.pUnsubscribe(ConvertUtil.toByteArray(this.stringRedisTemplate.getKeySerializer(), patterns));
            }else {
                subscription.pUnsubscribe();
            }
        }
    }

    /**
     * 获取订阅对象频道
     * @since redis 2.0.0
     * @return 返回频道列表
     */
    public List getChannelsAsObj() {
        Subscription subscription = this.redisTemplate.getRequiredConnectionFactory().getConnection().getSubscription();
        if (subscription!=null) {
            Collection<byte[]> channels = subscription.getChannels();
            return channels.parallelStream().map(bytes -> this.redisTemplate.getKeySerializer().deserialize(bytes)).collect(Collectors.toList());
        }
        return new ArrayList<>();
    }

    /**
     * 获取订阅字符串频道
     * @since redis 2.0.0
     * @return 返回频道列表
     */
    public List getChannels() {
        Subscription subscription = this.stringRedisTemplate.getRequiredConnectionFactory().getConnection().getSubscription();
        if (subscription!=null) {
            Collection<byte[]> channels = subscription.getChannels();
            return channels.parallelStream().map(bytes -> this.stringRedisTemplate.getKeySerializer().deserialize(bytes)).collect(Collectors.toList());
        }
        return new ArrayList<>();
    }

    /**
     * 获取spring redis模板
     * @return 返回对象模板
     */
    public RedisTemplate getRedisTemplate() {
        return this.redisTemplate;
    }

    /**
     * 获取spring string redis模板
     * @return 返回字符串模板
     */
    public StringRedisTemplate getStringRedisTemplate() {
        return this.stringRedisTemplate;
    }
}
